package com.share.tool.devops.controller;

import com.share.common.common.Result;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * kafka的启动与关闭
 * 当kafka在快速消费的时候调用stop或者start好像不好使
 * 20210611
 */
@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaController {
    @Autowired
    private KafkaListenerEndpointRegistry registry;

    /**
     * 断开当前消费者与kafka的连接
     *
     * @return
     */
    @RequestMapping("/stop")
    public Result stop() {
        KafkaMonitor kafkaMonitor = new KafkaMonitor();

        boolean updataBefore = registry.isRunning();
        long startTime = System.currentTimeMillis();
        log.info("Kafka listener 关闭 begin...");
        if (updataBefore) {
            registry.stop();
        }
        log.info("Kafka listener 关闭 end...");
        boolean updataAfter = registry.isRunning();
        long endTime = System.currentTimeMillis();

        kafkaMonitor.setCostTime(endTime - startTime);
        kafkaMonitor.setUpdataAfter(updataAfter);
        kafkaMonitor.setUpdataBefore(updataBefore);
        kafkaMonitor.setOper("stop");
        return Result.success(kafkaMonitor);
    }

    /**
     * 连接kafka
     *
     * @return
     */
    @RequestMapping("/start")
    public Result start() {
        KafkaMonitor kafkaMonitor = new KafkaMonitor();

        boolean updataBefore = registry.isRunning();
        long startTime = System.currentTimeMillis();
        log.info("Kafka listener 启动 begin...");
        if (!updataBefore) {
            registry.start();
        }
        log.info("Kafka listener 启动 end...");
        boolean updataAfter = registry.isRunning();
        long endTime = System.currentTimeMillis();

        kafkaMonitor.setCostTime(endTime - startTime);
        kafkaMonitor.setUpdataAfter(updataAfter);
        kafkaMonitor.setUpdataBefore(updataBefore);
        kafkaMonitor.setOper("start");
        return  Result.success(kafkaMonitor);
    }

    @RequestMapping("/status")
    public Result status() {
        boolean running = registry.isRunning();
        return  Result.success("kafka正在运行running=" + running);
    }

    @Data
    private class KafkaMonitor {
        private long costTime;
        private String oper;
        private boolean updataBefore;
        private boolean updataAfter;
    }
}
